package cn.doitedu.olap_agg;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/5/9
 * @Desc: 学大数据，上多易教育
 *  视频播放行为主题分析olap轻度聚合表计算
 *
 **/
public class Job03_VideoPlayOlapAggregate {

    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);


        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv.getConfig().set("table.exec.source.idle-timeout","1 ms");



        // 1. 创建kafka数据源表（行为日志明细宽表）
        tenv.executeSql(
                "create table dwd_events_kafka(     " +
                        "     event_id           string             " +
                        "    ,action_time        bigint             " +
                        "    ,properties         map<string,string> " +
                        "    ,video_id as properties['video_id'] " +
                        "    ,play_id as properties['play_id'] " +
                        "    ,user_id bigint             " +
                        "    ,province   string        " +
                        "    ,city       string        " +
                        "    ,region     string        " +
                        "    ,page_type       string   " +
                        "    ,page_service    string   " +
                        "    ,rt as to_timestamp_ltz(action_time,3)   " +
                        "    ,watermark for rt as rt - interval '0' second   " +
                        ") WITH (                      " +
                        "    'connector' = 'kafka',    " +
                        "    'topic' = 'dwd-user-action-log',  " +
                        "    'properties.bootstrap.servers' = 'doitedu:9092',  " +
                        "    'properties.group.id' = 'doit44_g1',              " +
                        "    'scan.startup.mode' = 'latest-offset',            " +
                        "    'value.format' = 'json',                          " +
                        "    'value.fields-include' = 'EXCEPT_KEY'             " +
                        ")                                                     ");


        tenv.executeSql("with tmp1 as (\n" +
                "    SELECT\n" +
                "       *\n" +
                "    FROM dwd_events_kafka\n" +
                "    WHERE event_id in ('video_play','video_hb','video_pause','video_resume','video_stop')\n" +
                ")\n" +
                ", tmp2 as (\n" +
                "    SELECT\n" +
                "    *,\n" +
                "    if(event_id = 'video_resume',1,0) as flag\n" +
                "    from tmp1  \n" +
                ")\n" +
                ", tmp3 as (\n" +
                "    SELECT\n" +
                "    *,\n" +
                "    sum(flag) over(partition by user_id,video_id,play_id order by  rt ) as flag2\n" +
                "    FROM tmp2 \n" +
                ")\n" +
                "\n" +
                ", tmp4 as (\n" +
                "SELECT\n" +
                "*,\n" +
                "first_value(action_time) over(partition by user_id,video_id,play_id,flag2 order by rt ) as part_start_time\n" +
                "from tmp3 \n" +
                ")\n" +
                "\n" +
                ", tmp5 as (\n" +
                "SELECT\n" +
                "    window_start,\n" +
                "    window_end,\n" +
                "    province,\n" +
                "    city,\n" +
                "    region,\n" +
                "    page_type,\n" +
                "    page_service,\n" +
                "    user_id,\n" +
                "    video_id,\n" +
                "    play_id,\n" +
                "    part_start_time,\n" +
                "    max(action_time) - part_start_time as part_time_long \n" +
                "\n" +
                "from table(\n" +
                "    tumble(table tmp4, descriptor(rt),interval '1' minute)\n" +
                ")\n" +
                "group by \n" +
                "    window_start,\n" +
                "    window_end,\n" +
                "    province,\n" +
                "    city,\n" +
                "    region,\n" +
                "    page_type,\n" +
                "    page_service,\n" +
                "    user_id,\n" +
                "    video_id,\n" +
                "    play_id,\n" +
                "    part_start_time\n" +
                ")\n" +
                "\n" +
                ",tmp6 as (\n" +
                "\n" +
                "SELECT\n" +
                "    province,\n" +
                "    city,\n" +
                "    region,\n" +
                "    page_type,\n" +
                "    page_service,\n" +
                "    user_id,\n" +
                "    video_id,\n" +
                "    play_id,\n" +
                "    part_start_time,\n" +
                "    max(part_time_long) as part_time_long\n" +
                "from tmp5\n" +
                "group by \n" +
                "    province,\n" +
                "    city,\n" +
                "    region,\n" +
                "    page_type,\n" +
                "    page_service,\n" +
                "    user_id,\n" +
                "    video_id,\n" +
                "    play_id,\n" +
                "    part_start_time\n" +
                "\n" +
                "\n" +
                "\n" +
                ")\n" +
                "\n" +
                "SELECT\n" +
                "    province,\n" +
                "    city,\n" +
                "    region,\n" +
                "    page_type,\n" +
                "    page_service,\n" +
                "    user_id,\n" +
                "    video_id,\n" +
                "    play_id,\n" +
                "    sum(part_time_long) as amt_time_long\n" +
                "from tmp5\n" +
                "group by \n" +
                "    province,\n" +
                "    city,\n" +
                "    region,\n" +
                "    page_type,\n" +
                "    page_service,\n" +
                "    user_id,\n" +
                "    video_id,\n" +
                "    play_id").print();



    }
}
